package com.xinqi.common.redis.delayed;

import com.xinqi.common.redis.delayed.api.DelayQueueHandler;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.redisson.spring.starter.RedissonAutoConfiguration;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.context.annotation.Bean;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 延时队列配置
 *
 * @author linkai
 * @date 2023/7/5
 */
@Slf4j
@AutoConfiguration
@AutoConfigureAfter({RedissonAutoConfiguration.class})
public class DelayQueueConfig {
    private final RedissonClient redissonClient;
    private final List<DelayQueueHandler> handlers;

    public DelayQueueConfig(ObjectProvider<RedissonClient> redissonClient, ObjectProvider<List<DelayQueueHandler>> handlers) {
        this.redissonClient = redissonClient.getIfAvailable();
        this.handlers = handlers.getIfAvailable();
    }

    @Bean
    public DelayQueueHelp delayQueueHelp() {
        return new DelayQueueHelp(redissonClient);
    }

    @Bean
    public CommandLineRunner delayQueueCommandLineRunner(DelayQueueHelp delayQueueHelp) {
        return args -> {
            if (handlers == null) {
                return;
            }
            // 根据 queueName 分组
            Map<String, List<DelayQueueHandler>> collect = handlers.stream().collect(Collectors.groupingBy(DelayQueueHandler::queueName));
            collect.forEach((queueName, queueHandlers) -> {
                Thread thread = new Thread(() -> {
                    while (true) {
                        String task = delayQueueHelp.getTask(queueName);
                        if (task != null) {
                            log.info("延时任务处理开始, task: {}, queueName: {}", task, queueName);
                            queueHandlers.forEach(queueHandler -> {
                                try {
                                    queueHandler.handleTask(task);
                                } catch (Exception e) {
                                    log.info("延时任务处理失败, task: {}, queueName: {}", task, queueName, e);
                                }
                            });
                        }
                    }
                }, "delay-queue-thread-" + queueName);
                thread.setDaemon(true);
                thread.start();
            });
        };
    }
}
